#!/usr/bin/env python

import os
import sys
from threading import Thread
from datetime import datetime

from pyfileutils import *
from pyfileutils.File import File
from pyfileutils.LockFile import LockFile

"""

A quick and dirty test script.

"""

FN = 'file_for_testing'
if os.path.isfile(FN):
    print >> sys.stderr, """
The testing script overwrites the file 'file_for_testing'.  If this is
important to you, you should back this file up or rename it.\n"""
    sys.exit(-1)

THREAD_NUM = 2
F = File(FN)
F.write('data')
START = datetime.now()
to_seconds = lambda x: x.days * 86400 + x.seconds
LF1 = LockFile(FN, overwrite=True)
THREAD_SAFETY_PASSED = True

class Checker(Thread):
    
    def run(self):
        global LF1
        global THREAD_SAFETY_PASSED
        while to_seconds(datetime.now() - START) <= 5: # do this for 5 seconds
            LF1.seek(0)
            LF1.write('lockeddata')
            contents = LF1.get_contents()
            try:
                assert contents == 'lockeddata'
            except AssertionError:
                THREAD_SAFETY_PASSED = False
                
def test1():
    print "Testing overwrite protection: ",
    try:
        g = File(FN)
        g.write('data')
        print "Failed"
    except: # this is what we want
        print "Passed"
    
def test2():
    print "Testing write consistency: ",
    try:
        assert F.get_contents() == 'data'
    except AssertionError:
        print "Failed initial write"
    F.append('moredata')
    try:
        assert F.get_contents() == 'datamoredata'
        print "Passed"
    except AssertionError:
        print "Failed append"
    
def test3():
    print "Testing delete: ",
    F.delete()
    try:
        assert not os.path.exists(FN)
        print "Passed"
    except AssertionError:
        print "Failed"
        
def test4():
    print "Testing LockFile equality: ",
    lf2 = LockFile(FN, overwrite=True)
    try:
        assert LF1 == lf2
        print "Passed"
    except AssertionError:
        print "Failed"
    
def test5():
    global THREAD_SAFETY_PASSED
    print "Testing LockFile threadsafety: ",
    threads = [Checker() for x in range(THREAD_NUM)]
    for t in threads: t.start()
    for t in threads: t.join()
    if THREAD_SAFETY_PASSED:
        print "Passed"
    else:
        print "Failed"
        
for t in [test1, test2, test3, test4, test5]:
    t()
    
try:
    os.unlink(FN)
except Exception, e:
    print >> sys.stderr, "Error removing %s: [%s]" % (FN, str(e))
